﻿using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using Service;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace Service
{
    //基于http://rabbitmq.mr-ping.com/tutorials_with_csharp/

    //direct(直射)、topic(主题)、fanout(扩散)、headers(规则外)四种交换模式的区别：
    //direct,将所有的发送到交换器的消息完全按照BindingKey和RoutingKey匹配，再转发到相应的queue。
    //topic,通过消息的Routing Key和Exchange、Queue进行匹配，将消息路由给一个或多个队列，达到发布/订阅的效果。
    //Fanout,不处理路由键会把所有发送到交换器的消息转发到所有绑定的队列中。优点是转发消息最快，性能最好。
    //headers,该类型的交换器不依赖路由规则来路由消息，而是根据消息内容中的headers属性进行匹配。headers类型交换器性能差，在实际中并不常用。
    public enum ExchangeTypes
    {
        direct,
        fanout,
        headers,
        topic
    }


    public class MessageQueue
    {
        ConnectionFactory connectionFactory;
        List<Action<IModel>> CommandQueue;

        public MessageQueue(string HostName = "localhost", int Port = 5672, string UserName = "admin", string Password = "admin")
        {
            connectionFactory = new ConnectionFactory()
            {
                HostName = HostName,
                Port = Port,
                UserName = UserName,
                Password = Password
            };
            CommandQueue = new List<Action<IModel>>();
        }

        public MessageQueue(ConnectionFactory _connectionFactory)
        {
            connectionFactory = _connectionFactory;
            CommandQueue = new List<Action<IModel>>();
        }


        private void BuildConnectAndCannle(Action<IModel> exec = null)
        {
            //创建连接
            using (var connect = connectionFactory.CreateConnection())
            {
                //创建频道
                using (var channle = connect.CreateModel())
                {
                    exec?.Invoke(channle);
                }
            }
        }

        private IConnection BuildConnect()
        {
            //创建连接
            return connectionFactory.CreateConnection();
        }

        /// <summary>
        /// 创建队列
        /// </summary>
        /// <param name="queueName">队列名</param>
        /// 
        /// <param name="durable">持久化</param>
        /// 
        /// <param name="exclusive">排外的</param>
        ///                         是否排外的，有两个作用:
        ///                         一：当连接关闭时connection.close()该队列是否会自动删除；
        ///                         二：该队列是否是私有的private，如果不是排外的，可以使用两个消费者都访问同一个队列，没有任何问题，如果是排外的，会对当前队列加锁，其他通道channel是不能访问的，如果强制访问会报异常
        ///                         
        /// <param name="autoDelete">自动删除</param>
        ///                          是否自动删除，当最后一个消费者断开连接之后队列是否自动被删除
        /// 
        /// <param name="arguments">参数</param>
        ///                         建议查询文档
        ///                         
        public void CreateQueue(string queueName, bool durable = false, bool exclusive = false, bool autoDelete = true, IDictionary<string, object> arguments = null)
        {
            CommandQueue.Add(channel =>
            {
                channel.QueueDeclare(queueName, durable, exclusive, autoDelete, arguments);
            });
        }

        public void CreateExchange(string exchangeName, ExchangeTypes exchangeType, bool durable = false, bool autoDelete = true, IDictionary<string, object> arguments = null)
        {
            CommandQueue.Add(channel =>
            {
                channel.ExchangeDeclare(exchangeName,exchangeType.ToString(), durable, autoDelete, arguments);
            });
        }

        public void BindQueue(string queueName,string exchangeName,string routingKey, IDictionary<string, object> arguments = null)
        {
            CommandQueue.Add(channel =>
            {
                channel.QueueBind(queueName,exchangeName, routingKey, arguments);
            });
        }

        public void SendMessage(string routingKey,string exchangeName,string Message, Func<IBasicProperties, IBasicProperties>  basicProperties = null)
        {
            CommandQueue.Add(channel =>
            {
                var body = Encoding.UTF8.GetBytes(Message);

                var properties = basicProperties?.Invoke(channel.CreateBasicProperties());
                channel.BasicPublish(exchangeName, routingKey, properties, body);
            });
        }

        /// <summary>
        /// 创建消费者
        /// </summary>
        /// <param name="queueName">监听队列名</param>
        /// <param name="exec">监听消息事件</param>
        /// <param name="autoAck">自动答应</param>    当消费者受到消息就进行表明，之后队列会把此消息删除
        /// <returns></returns>
        public IConnection GetMessage(string queueName, Action<object, BasicDeliverEventArgs> exec, bool autoAck = false)
        {
            var connect = BuildConnect();
            var channel = connect.CreateModel();
            var consumer = new EventingBasicConsumer(channel);
            consumer.Received += (model, ea) =>
            {
                exec?.Invoke(model, ea);
            };
            channel.BasicConsume(queueName, autoAck, consumer);
            return connect;
        }

        public IConnection GetMessage(string queueName, Action<object, BasicDeliverEventArgs,IModel> exec, bool autoAck = false)
        {
            var connect = BuildConnect();
            var channel = connect.CreateModel();
            var consumer = new EventingBasicConsumer(channel);
            consumer.Received += (model, ea) =>
            {
                exec?.Invoke(model, ea,channel);
            };
            channel.BasicConsume(queueName, autoAck, consumer);
            return connect;
        }

        public void Execute()
        {
            var connect = BuildConnect();
            var channel = connect.CreateModel();
            CommandQueue.ForEach(item =>
            {
                item.Invoke(channel);
            });
            connect.Close();
            CommandQueue.Clear();
        }

        public void Execute(Func<IModel, IModel> ChannelSet)
        {
            var connect =  BuildConnect();
            var channel = ChannelSet?.Invoke(connect.CreateModel());
            CommandQueue.ForEach(item =>
            {
                item.Invoke(channel);
            });
            connect.Close();
            CommandQueue.Clear();
        }

    }
}